package com.xujianlong.combat;

import com.xujianlong.bean.ItemCount;
import com.xujianlong.bean.UserBehavior;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.concurrent.ThreadPoolExecutor;

public class FLink_03_Project_ItemCountTopN {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStreamSource<String> source = env.readTextFile("input/UserBehavior.csv");

        SingleOutputStreamOperator<UserBehavior> userBehaviorWithWMDS = source.map(data -> {
            String[] split = data.split(",");
            UserBehavior userBehavior = new UserBehavior(Long.parseLong(split[0]),
                    Long.parseLong(split[1]),
                    Integer.parseInt(split[2]),
                    split[3],
                    Long.parseLong(split[4]));
            return userBehavior;
        }).filter(data->"pv".equals(data.getBehavior())).assignTimestampsAndWatermarks(WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
            @Override
            public long extractTimestamp(UserBehavior userBehavior, long l) {
                return userBehavior.getTimestamp() * 1000;
            }
        }));
//        KeyedStream<UserBehavior, Long> keyedDS = userBehaviorWithWMDS.keyBy(data -> data.getItemId());
        KeyedStream<Tuple2<Long, Integer>, Long> keyedDS = userBehaviorWithWMDS.map(new MapFunction<UserBehavior, Tuple2<Long, Integer>>() {
            @Override
            public Tuple2<Long, Integer> map(UserBehavior userBehavior) throws Exception {
                return Tuple2.of(userBehavior.getItemId(), 1);
            }
        }).keyBy(data -> data.f0);
        SingleOutputStreamOperator<ItemCount> aggregate = keyedDS.window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .aggregate(new ItemCountAggFunc(), new ItemCountWindowFunc());

        KeyedStream<ItemCount, String> itemCountStringKeyedStream = aggregate.keyBy(ItemCount::getTime);
        SingleOutputStreamOperator<String> process = itemCountStringKeyedStream.process(new ItemCountProcessFunc(5));

        process.print();

        env.execute();
    }
    public static class ItemCountAggFunc implements AggregateFunction<Tuple2<Long, Integer>,Integer,Integer>{

        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(Tuple2<Long, Integer> longIntegerTuple2, Integer integer) {
            return integer+1;
        }

        @Override
        public Integer getResult(Integer integer) {
            return integer;
        }

        @Override
        public Integer merge(Integer integer, Integer acc1) {
            return integer+acc1;
        }
    }

    public static class ItemCountWindowFunc implements WindowFunction<Integer, ItemCount, Long, TimeWindow> {

        @Override
        public void apply(Long aLong, TimeWindow window, Iterable<Integer> input, Collector<ItemCount> out) throws Exception {
            Integer count = input.iterator().next();
            out.collect(new ItemCount(aLong,new Timestamp(window.getEnd()).toString(),count));
        }
    }
    public static class ItemCountProcessFunc extends KeyedProcessFunction<String,ItemCount,String>{
        private Integer topSize;

        private ListState<ItemCount> listState;
        public ItemCountProcessFunc(Integer n){
            topSize=n;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            listState=getRuntimeContext().getListState(new ListStateDescriptor<ItemCount>("list-state",ItemCount.class));
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            Iterable<ItemCount> itemCounts = listState.get();
            ArrayList<ItemCount> itemCounts1 = Lists.newArrayList(itemCounts.iterator());

            itemCounts1.sort((o1,o2)->o2.getCount()-o1.getCount());
            StringBuilder sb = new StringBuilder();
            sb.append("========").append(new Timestamp(timestamp-1000L))
                    .append("========")
                    .append("\n");
            for (int i=0;i<Math.min(topSize,itemCounts1.size());i++){
                ItemCount itemCount = itemCounts1.get(i);
                sb.append("Top").append(i+1);
                sb.append(" ItemId:"+itemCount.getItem());
                sb.append(" Count:").append(itemCount.getCount());
                sb.append("\n");
            }
            sb.append("========").append(new Timestamp(timestamp-1000L))
                    .append("========")
                    .append("\n")
                    .append("\n");

            listState.clear();
            out.collect(sb.toString());

            Thread.sleep(2000);
        }

        @Override
        public void processElement(ItemCount value, Context ctx, Collector<String> out) throws Exception {
            listState.add(value);

            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            ctx.timerService().registerEventTimeTimer(sdf.parse(value.getTime()).getTime()+1000L);

        }
    }

}
